package club.gongsir.hbase;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

/**
 * @author 龚涛
 * @date 2020/7/4 18:44
 * 编码不要畏惧变化，要拥抱变化
 */
public class HbaseTest {
    private static Configuration conf = HBaseConfiguration.create();
    private static Connection conn;

    public static void main(String[] args) {
        //创建HBase配置
        conf.set("hbase.rootdir", "hdfs://node1:9000/hbase");
        conf.set("hbase.zookeeper.quorum", "node1,node2,node3");
        conf.set("hbase.zookeeper.property.clientPort", "2181");
        try {
            conn = ConnectionFactory.createConnection(conf);
            System.out.println("向数据库插入indexes数据:");
            putSingleValue("Student","index","Base","indexes","001,002,004");
            String indexes = getSingleValue("Student", "index", "Base", "indexes");
            System.out.println("获取indexes数据:" + indexes);
            assert indexes != null;
            String[] idx = indexes.split(",");
            for (String index :
                    idx ) {
                System.out.println(getMsgForLab7(index));
            }
            conn.close();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 创建表
     * @param tabName 表名
     * @param columnFamilies 列族集合
     * @throws IOException
     */
    private static void createTable(String tabName,String[] columnFamilies) throws IOException {
        //构造表名对象
        TableName tableName = TableName.valueOf(tabName);
        //获取管理对象
        Admin admin = conn.getAdmin();
        //判断表是否存在
        if (!admin.tableExists(tableName)) {
            //构造表描述器器
            TableDescriptorBuilder tableDescriptor = TableDescriptorBuilder.newBuilder(tableName);
            for (String columnFamily:
                 columnFamilies) {
                ColumnFamilyDescriptor columnFamilyDescriptor = ColumnFamilyDescriptorBuilder.newBuilder(Bytes.toBytes(columnFamily)).build();
                tableDescriptor.setColumnFamily(columnFamilyDescriptor);
            }
            //获得表描述器
            TableDescriptor td = tableDescriptor.build();
            //创建表
            admin.createTable(td);
            System.out.println("create table " + tableName.getNameAsString() + "success");
        } else {
            System.out.println("table " + tableName.getNameAsString() + "exists");
        }
    }

    /**
     * 新增单条数据
     * @param tableStr 表名
     * @param rowKey 行键
     * @param columnFamily 列族
     * @param columnName 列名
     * @param cellValue cell值
     * @throws IOException
     */
    private static void putSingleValue(String tableStr, String rowKey, String columnFamily, String columnName, String cellValue) throws IOException{
        TableName tableName = TableName.valueOf(tableStr);
        //Table负责跟记录相关的操作如增删改查等
        Table table = conn.getTable(tableName);
        Put put=new Put(Bytes.toBytes(rowKey));
        put.addColumn(Bytes.toBytes(columnFamily), Bytes.toBytes(columnName), Bytes.toBytes(cellValue));
        table.put(put);
        table.close();
        System.out.println("add data Success!");
    }

    /**
     * 删除单条数据
     * @param tableStr 表名
     * @param rowKey 行键
     * @param columnFamily 列族
     * @param columnName 列名
     * @throws IOException
     */
    private static void deleteSingleValue(String tableStr,String rowKey, String columnFamily, String columnName) throws IOException {
        TableName tableName = TableName.valueOf(tableStr);
        //创建表对象，实现表删除
        Table table = conn.getTable(tableName);
        //定义行
        Delete delete = new Delete(Bytes.toBytes(rowKey));
        //delete.addFamily(Bytes.toBytes(columnFamily));    //定义需要删除的列族
        //delete.addColumn(Bytes.toBytes(columnFamily), Bytes.toBytes(columnName));   //定义需要删除的列
        table.delete(delete);
        table.close();
    }

    /**
     * 删除表
     * @param tableStr 表名
     * @throws IOException
     */
    private static  void deleteTable(String tableStr) throws IOException{
        TableName tableName = TableName.valueOf(tableStr);
        //创建管理对象并禁用表，删除表前先禁用掉
        Admin admin = conn.getAdmin();
        admin.disableTable(tableName);
        admin.deleteTable(tableName);
    }

    /**
     * 获取单条数据
     * @param tableStr 表名
     * @param rowKey 行键
     * @param columnFamily 列族
     * @param columnName 列名
     * @throws IOException
     */
    private static String getSingleValue(String tableStr, String rowKey, String columnFamily, String columnName) throws IOException {
        List<String> values = new ArrayList<>();
        TableName tableName = TableName.valueOf(tableStr);
        Table table = conn.getTable(tableName);
        Get get = new Get(Bytes.toBytes(rowKey));
        get.addColumn(Bytes.toBytes(columnFamily), Bytes.toBytes(columnName));
        //执行数据读取并返回结果对象
        Result set = table.get(get);
        //获取一行数据集中的所有单元格(cell)对象
        Cell[] cells  = set.rawCells();
        //遍历所有单元格对象
        for(Cell cell : cells) {
            byte[] cellValue = cell.getValueArray();
            values.add(Bytes.toString(cellValue, cell.getValueOffset(), cell.getValueLength()));
        }
        table.close();
        if (!values.isEmpty()) {
            return values.get(0);
        } else {
            return "";
        }
    }

    private static String getMsgForLab7(String index) throws IOException {
        return getSingleValue("Student", index, "Base", "Name") +
                "(" +
                getSingleValue("Student", index, "Base", "Number") +
                "):" +
                getSingleValue("Student", index, "Score", "English") +
                " " +
                getSingleValue("Student", index, "Score", "Java");
    }
}
